package com.zws.cucumber.glue;

import com.datastax.oss.driver.api.core.cql.BatchableStatement;
import com.datastax.oss.driver.api.core.cql.BoundStatement;
import com.datastax.oss.driver.api.core.cql.PreparedStatement;
import com.datastax.oss.driver.api.core.cql.Statement;
import com.datastax.oss.driver.api.core.metadata.schema.TableMetadata;
import com.datastax.oss.driver.api.core.type.DataType;
import com.datastax.oss.driver.api.core.type.UserDefinedType;
import com.datastax.oss.driver.internal.core.data.DefaultUdtValue;
import com.datastax.oss.driver.internal.core.util.Strings;
import com.zws.cucumber.adaptation.cass.ZwsCqlDmlType;
import com.zws.cucumber.adaptation.cass.ZwsCqlSessionProxy;
import com.zws.cucumber.adaptation.cass.ZwsCqlSessionProxyHelper;
import com.zws.cucumber.bootstrap.ZwsCassBootstrap;
import com.zws.cucumber.expr.ZwsExpr;
import com.zws.cucumber.expr.ZwsExprCtx;
import com.zws.cucumber.spring.configuration.ZwsCassConf;
import com.zws.cucumber.util.cass.ZwsCassContext;
import com.zws.cucumber.util.cass.ZwsCassTemplate;
import com.zws.cucumber.util.cass.ZwsCassTemplate.ColumnMetadata;
import com.zws.cucumber.util.cass.ZwsCassTemplate.ColumnType;
import com.zws.cucumber.util.colorfullog.ZwsColorfulLogger;
import com.zws.cucumber.util.colorfullog.ZwsPrintTableHelper;
import com.zws.cucumber.var.ZwsVar;
import io.cucumber.datatable.DataTable;
import io.cucumber.java.After;
import io.cucumber.java.Before;
import io.cucumber.java.en.Given;
import io.cucumber.java.en.Then;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.junit.jupiter.api.Assertions;
import org.slf4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;

import java.util.*;
import java.util.stream.Collectors;

import static com.zws.cucumber.expander.ZwsStr.expand;
import static com.zws.cucumber.expr.ZwsExpr.evaluate;

/**
 * author: zws
 */
public class ZwsCassStepDef {

    private static final Logger logger = ZwsColorfulLogger.of(ZwsCassStepDef.class);

    @Autowired
    private ZwsCassTemplate template;

    @Autowired
    private ZwsCassConf.TestSessionFactory sessionFactory;

    @Value("${ct.cucumber.cassandra.clearEachCase:true}")
    private boolean clearCassandraEachCase;

    @Value("${ct.keyspace.name}")
    private String keyspace;

    @Before
    public void cleanData() {
        if (clearCassandraEachCase) {
            String keyspaceToClear = ZwsCassContext.getGeneratedKeySpace(keyspace);
            logger.debug("clearCassandraEachCase: {}, keyspace {}", clearCassandraEachCase,
                keyspaceToClear);
            ZwsCqlSessionProxy.asyncTruncate(template, keyspaceToClear);
        }
    }

    @Deprecated
    @Before("@BeforeDatabaseException")
    public void beforeDatabaseException() {
        // until now, just support cassandra
        ZwsCqlSessionProxy.enableThrowingException();
    }

    @Deprecated
    @After("@AfterDatabaseException")
    public void afterDatabaseException() {
        // until now, just support cassandra
        ZwsCqlSessionProxy.disableThrowingException();
    }

    @Deprecated
    @Given("database crash")
    @Before("@StartWithCassandraCrash")
    public void databaseCrash() {
        // until now, just support cassandra
        ZwsCqlSessionProxy.enableThrowingException();
    }

    @Deprecated
    @Given("database resume")
    @After("@EndWithCassandraResume")
    public void databaseResume() {
        // until now, just support cassandra
        ZwsCqlSessionProxy.disableThrowingException();
    }

    /*
     * exception in cassandra driver
     */
    @Given("enable Cassandra Driver throwing {word} exception for table {string} in {string} statement")
    public void enableCassandraDriverThrowingException(String ex, String table, String dmlType) {
        ZwsCqlSessionProxy.enableThrowingException(
                switch (ex.toLowerCase()) {
                    case "timeout" -> ZwsCqlSessionProxy.EX_TIMEOUT;
                    case "no_node_available" -> ZwsCqlSessionProxy.EX_NO_NODE_AVAILABLE;
                    default -> throw new IllegalArgumentException();
                },
                table,
                ZwsCqlDmlType.of(dmlType)
        );
    }

    @Given("enable Cassandra Driver throwing {word} exception for table {string}")
    public void enableCassandraDriverThrowingException(String ex, String table) {
        enableCassandraDriverThrowingException(ex, table, null);
    }

    @Given("enable Cassandra Driver throwing {word} exception in {string} statement")
    public void enableCassandraDriverThrowingExceptionInStatement(String ex, String dmlType) {
        enableCassandraDriverThrowingException(ex, null, dmlType);
    }

    @Given("enable Cassandra Driver throwing {word} exception")
    public void enableCassandraDriverThrowingException(String ex) {
        enableCassandraDriverThrowingException(ex, null, null);
    }

    @Before("@StartWithEnablingCassandraDriverThrowingException")
    @Given("enable Cassandra Driver throwing exception")
    public void enableCassandraDriverThrowingException() {
        enableCassandraDriverThrowingException("timeout", null, null);
    }

    @After("@EndWithDisableCassandraDriverThrowingException")
    @Given("disable Cassandra Driver throwing exception")
    public void disableCassandraDriverThrowingException() {
        ZwsCqlSessionProxy.disableThrowingException();
    }

    /*
     * timeout in cassandra server
     *
     * NOTES: ONLY RUN IN WINDOWS
     */
    @Given("enable Cassandra timeout for table {string} in {string} statement")
    public void enableCassandraTimeout(String targetTable, String dataManipulationType) {
        if (ZwsCassBootstrap.cassBackdoorClient != null) {
            ZwsCqlSessionProxy.asyncExecute(Collections.emptyList(), () -> ZwsCassBootstrap.cassBackdoorClient.enableTimeout(dataManipulationType, targetTable));
        } else {
            throw new RuntimeException("Not support this step");
        }
    }

    @Given("enable Cassandra timeout for table {string}")
    public void enableCassandraTimeout(String targetTable) {
        enableCassandraTimeout(targetTable, null);
    }

    @Given("enable Cassandra timeout in {string} statement")
    public void enableCassandraTimeoutInStatement(String dataManipulationType) {
        enableCassandraTimeout(null, dataManipulationType);
    }

    @Given("enable Cassandra timeout")
    public void enableCassandraTimeout() {
        enableCassandraTimeout(null, null);
    }

    @Given("disable Cassandra timeout")
    @After("@EndWithDisablingCassandraTimeout")
    public void disableCassandraTimeout() {
        if (ZwsCassBootstrap.cassBackdoorClient != null) {
            ZwsCqlSessionProxy.asyncExecute(Collections.emptyList(), () -> ZwsCassBootstrap.cassBackdoorClient.disableTimeout());
        } else {
            throw new RuntimeException("Not support this step");
        }
    }

    /*
     *
     */
    @Then("execute ql")
    public void executeQl(String ql) {
        expandAndExecute(ql);
    }

    @Then("execute ql {string}")
    public void executeSimpleQl(String ql) {
        executeQl(ql);
    }

    @Given("insert into table {string}")
    public void insertIntoTable(String tableName, DataTable dataTable) {

        tableName = tableName.toLowerCase();

        List<String> columnList = dataTable.row(0)
                .stream()
                .map(String::toLowerCase)
                .collect(Collectors.toList());

        List<List<String>> cells = new ArrayList<>(dataTable.cells());
        cells.set(0, columnList);
        dataTable = DataTable.create(cells, dataTable.getTableConverter());

        TableMetadata tableMetadata = ZwsCqlSessionProxyHelper.getTableMetadata(tableName);
        PreparedStatement preparedStmt = ZwsCqlSessionProxyHelper.getInsertionPreparedStatement(sessionFactory.getSession(), tableName, columnList);

        logger.debug("dataTable.size: {}", dataTable.cells().size());
        List<BoundStatement> stmtList = new ArrayList<>();
        for (Map<String, String> row : dataTable.asMaps()) {
            Map<String, String> expandedAndEvaluatedRow = new HashMap<>();
            row.forEach((k, v) -> expandedAndEvaluatedRow.put(k, evaluate(k, expand(v))));

            BoundStatement stmt = ZwsCqlSessionProxyHelper.generateBoundStatement(tableMetadata, preparedStmt, expandedAndEvaluatedRow);
            stmtList.add(stmt);
        }
        if (stmtList.size() == 1) {
            insert0(tableName, stmtList.get(0));
        } else if (stmtList.size() > 1) {
            insertBatch0(tableName, stmtList);
        }
    }

    @Given("insert single row into table {string}")
    public void insertSingleRowIntoTable(String tableName, DataTable dataTable) {
        List<List<String>> originTable = dataTable.cells();
        List<String> title = new ArrayList<>();
        List<String> row = new ArrayList<>();
        for (List<String> originRow : originTable) {
            title.add(originRow.get(0));
            row.add(originRow.get(1));
        }
        insertIntoTable(tableName, DataTable.create(Arrays.asList(title, row), dataTable.getTableConverter()));
    }

    @Given("verify data in table {string}")
    public void verifyDataInTable(String tableName, DataTable dataTable) {
        verifyDataInTableWithWhereClause(tableName, null, dataTable);
    }

    @Given("verify data in table {string} where {string}")
    public void verifyDataInTableWithWhereClause(String tableName, String whereClause, DataTable dataTable) {
        ZwsCqlSessionProxy.waitAsyncExecution();
        List<String> titleList = dataTable.row(0);
        List<String> columnList = new ArrayList<>();
        List<String> originalColumnList = new ArrayList<>();
        Map<String, String> column2PrefixMap = new HashMap<>();
        for (int i = 0; i < titleList.size(); i++) {
            String title = titleList.get(i);
            String[] array = title.split(":");
            String originalColumn = title;
            String column = null;
            if (array.length == 1) {
                column = normalizeColumnName(originalColumn);
            } else if (array.length == 2) {
                originalColumn = array[0].trim();
                column = normalizeColumnName(originalColumn);
                String prefix = array[1].trim();
                column2PrefixMap.put(column, prefix);
            }
            originalColumnList.add(originalColumn);
            columnList.add(column);
        }
        String columnStmt = String.join(",", originalColumnList);

        String ql = null;
        if (StringUtils.isNotBlank(whereClause)) {
            ql = String.join(" ", "select", columnStmt, "from", tableName, "where", whereClause);
        } else {
            ql = String.join(" ", "select", columnStmt, "from", tableName);
        }
        logger.debug("ql: {}", ql);
        Pair<Map<String, ColumnMetadata>, List<Map<String, Object>>> metadataAndActualRows = expandAndQueryWithMetadata(ql);
        List<Map<String, Object>> actualRows = metadataAndActualRows.getRight();
        List<Map<String, Object>> duplicatedActualRows = new ArrayList<>(actualRows);
        List<Map<String, String>> expectedRows = normalizeColumnName(dataTable.asMaps());
        ZwsPrintTableHelper.printTable0(
                tableName,
                Pair.of(new ArrayList<>(metadataAndActualRows.getLeft().keySet()), metadataAndActualRows.getRight()),
                true);
        Assertions.assertEquals(expectedRows.size(), actualRows.size());

        TableMetadata tableMetadata = getTableMetadata(tableName);
        Set<Integer> matchedIndexes = new HashSet<>();
        for (int i = 0; i < expectedRows.size(); i++) {
            Map<String, String> expectedRow = expectedRows.get(i);
            logger.debug("try to match expected row {}", i);
            boolean hasSame = false;
            final int idx = i;
            for (int j = 0; j < duplicatedActualRows.size(); j++) {
                if (matchedIndexes.contains(j)) {
                    continue;
                }
                logger.debug("<c>----------------try to compare expect row {} to actual row {}</c>", i, j);
                Map<String, Object> actualRow = duplicatedActualRows.get(j);
                if (isRowEquals(expectedRow, actualRow, columnList, tableMetadata)) {
                    logger.debug("<g><bold>----------------expect row {} match with actual row {}</bold></g>", i, j);
//                    duplicatedActualRows.remove(j);
                    matchedIndexes.add(j);
                    hasSame = true;
                    column2PrefixMap.forEach((column, prefix) -> ZwsVar.put(prefix + (idx + 1), actualRow.get(column)));
                    break;
                } else {
                    logger.debug("<y><bold>----------------expect row {} doesn't match with actual row {}</bold></y>", i, j);
                }
            }
            if (!hasSame) {
                Assertions.fail("no row equals, expected row: " + expectedRow + ", actual row list: " + actualRows);
            }
        }
    }

    @Given("verify single row in table {string} where {string}")
    public void verifySingleRowInTableWithWhereClause(String tableName, String whereClause, DataTable dataTable) {
        List<List<String>> originTable = dataTable.cells();
        List<String> title = new ArrayList<>();
        List<String> row = new ArrayList<>();
        for (List<String> originRow : originTable) {
            title.add(originRow.get(0));
            row.add(originRow.get(1));
        }
        verifyDataInTableWithWhereClause(tableName, whereClause, DataTable.create(Arrays.asList(title, row), dataTable.getTableConverter()));
    }

    @Given("verify no data in table {string}")
    public void verifyNoDataInTable(String tableName) {
        verifyNoDataInTableWithWhereClause(tableName, null);
    }

    @Given("verify no data in table {string} where {string}")
    public void verifyNoDataInTableWithWhereClause(String tableName, String whereClause) {
        ZwsCqlSessionProxy.waitAsyncExecution();
        String ql;
        if (StringUtils.isNotBlank(whereClause)) {
            ql = String.join(" ", "select", "*", "from", tableName, "where", whereClause);
        } else {
            ql = String.join(" ", "select", "*", "from", tableName);
        }
        logger.debug("ql: {}", ql);
        List<Map<String, Object>> actualRows = expandAndQuery(ql);
        logger.debug("result: {}", actualRows);
        Assertions.assertEquals(0, actualRows.size());
    }

    @Given("print table {string}")
    public void printTable(String tableName) {
        printTableWithWhereClause(tableName, null);
    }

    @Given("print table {string} where {string}")
    public void printTableWithWhereClause(String tableName, String whereClause) {
        ZwsCqlSessionProxy.waitAsyncExecution();
        String ql;
        if (StringUtils.isNotBlank(whereClause)) {
            ql = String.join(" ", "select", "*", "from", tableName, "where", whereClause);
        } else {
            ql = String.join(" ", "select", "*", "from", tableName);
        }
        logger.debug("ql: {}", ql);
        Pair<Map<String, ColumnMetadata>, List<Map<String, Object>>> metadataAndActualRows = expandAndQueryWithMetadata(ql);
        logger.trace("query result: {}", metadataAndActualRows);
        ZwsPrintTableHelper
            .printTable0(tableName, Pair.of(new ArrayList<>(metadataAndActualRows.getLeft().keySet()), metadataAndActualRows.getRight()));
    }

    private List<Map<String, Object>> expandAndQuery(String ql) {
        return normalizeColumnName(query(expand(ql)));
    }

    private Pair<Map<String, ColumnMetadata>, List<Map<String, Object>>> expandAndQueryWithMetadata(String ql) {
        return queryWithMetadata(expand(ql));
    }

    private List<Map<String, Object>> query(String ql) {
        ZwsCqlSessionProxy.waitAsyncExecution();
        return normalizeColumnName(template.queryList(ql));
    }

    private Pair<Map<String, ColumnMetadata>, List<Map<String, Object>>> queryWithMetadata(String ql) {
        ZwsCqlSessionProxy.waitAsyncExecution();
        Pair<Map<String, ColumnMetadata>, List<Map<String, Object>>> pair = template.queryListWithMetadata(ql);
        return Pair.of(pair.getKey(), normalizeColumnName(pair.getValue()));
    }

    private void insert(String modifiedTable, String cql) {
        ZwsCqlSessionProxy.asyncExecute(modifiedTable, cql);
    }

    private void insert0(String modifiedTable, Statement stmt) {
        ZwsCqlSessionProxy.asyncExecute(Collections.singletonList(modifiedTable), stmt);
    }

    private void insertBatch(String modifiedTable, List<String> cqlList) {
        ZwsCqlSessionProxy.asyncExecute(Collections.singletonList(modifiedTable), cqlList);
    }

    private void insertBatch0(String modifiedTable, List<? extends BatchableStatement> batchableStatementList) {
        ZwsCqlSessionProxy.asyncExecute0(Collections.singletonList(modifiedTable), batchableStatementList);
    }

    private void expandAndExecute(String ql) {
        ZwsCqlSessionProxy.waitAsyncExecution();
        template.execute(expand(ql));
    }

    private boolean isRowEquals(Map<String, String> expectedRow, Map<String, Object> actualRow, List<String> columnList, TableMetadata tableMetadata) {
        for (String column : columnList) {
            // get expected field
            String expectedField = expectedRow.get(column);
            // get and normalize actual field
            Object actualField = actualRow.get(column);
            ZwsExprCtx ctx = initCucumberExprContext(getColumnDataType(tableMetadata, column));
            actualField = normalizeValue(actualField);

            if (!ZwsExpr.verifySilently(column, expand(expectedField), actualField, ctx)) {
                return false;
            }
        }
        return true;
    }

    private static Object normalizeValue(Object value) {
        if (value instanceof Number) {
            return value;
        }
        if (value instanceof DefaultUdtValue) {
            DefaultUdtValue udtValue = ((DefaultUdtValue) value);
            UserDefinedType udt = udtValue.getType();
            Map<String, Object> newMap = new LinkedHashMap<>();
            for (int i = 0; i < udt.getFieldTypes().size(); i++) {
                newMap.put(udt.getFieldNames().get(i).asInternal(), normalizeValue(udtValue.getObject(i)));
            }
            return newMap;
        }
        if (value instanceof Map) {
            Map<String, Object> newMap = new LinkedHashMap<>();
            ((Map) value).forEach((k, v) -> {
                newMap.put((String) normalizeValue(k), normalizeValue(v));
            });
            return newMap;
        }
        if (value instanceof Collection) {
            return ((Collection) value).stream()
                .map(ZwsCassStepDef::normalizeValue)
                .collect(Collectors.toList());
        }
        return value;
    }

    private String convert(Map<String, ColumnMetadata> metadataMap, String column, String value) {
        ColumnType type = metadataMap.get(column.toLowerCase()).getType();
        switch (type) {
            case STR:
                return value != null ? "'" + value + "'" : null;
            default:
                return value;
        }
    }

    private String parentheses(String text) {
        return "(" + text + ")";
    }

    private <V> List<Map<String, V>> normalizeColumnName(List<Map<String, V>> originList) {
        List<Map<String, V>> resultList = new ArrayList<>();
        for (Map<String, V> origin : originList) {
            Map<String, V> result = new LinkedHashMap<>();
            origin.forEach((k, v) -> {
                if (k.contains(":")) {
                    result.put(k.split(":")[0].trim().toLowerCase(), v);
                } else {
                    result.put(k.trim().toLowerCase(), v);
                }
            });
            resultList.add(result);
        }
        return resultList;
    }

    private String normalizeColumnName(String columnName) {
        if (columnName == null) {
            return null;
        }
        if (columnName.contains(":")) {
            return columnName.split(":")[0].trim().toLowerCase();
        } else {
            return columnName.toLowerCase();
        }
    }

    private ZwsExprCtx initCucumberExprContext(DataType dataType) {
        ZwsExprCtx ctx = new ZwsExprCtx();
        ctx.setDataType(dataType);
        return ctx;
    }

    private TableMetadata getTableMetadata(String tableName) {
        logger.debug("keyspaces: {}", sessionFactory.getSession().getMetadata().getKeyspaces());
        return sessionFactory.getSession()
                .getMetadata()
                .getKeyspace(ZwsCassContext.getGeneratedKeySpace(keyspace)).get()
                .getTable(tableName).get();
    }

    private DataType getColumnDataType(TableMetadata tableMetadata, String columnName) {
//        logger.debug("columns: {}", tableMetadata.getColumns());
        com.datastax.oss.driver.api.core.metadata.schema.ColumnMetadata columnMetadata = tableMetadata
                .getColumn(Strings.needsDoubleQuotes(columnName) ? Strings.doubleQuote(columnName) : columnName)
                .orElse(null);
        DataType type = null;
        if (columnMetadata != null) {
            type = columnMetadata.getType();
        }
        return type;
    }

}
